package com.codejiwei.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/6/8
 * desc:
 **/
public class Flink_Table_SQL_API4 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //source
        tableEnv.executeSql("create temporary table sourceTable(col1 String, col2 bigint, col3 timestamp) " +
                "with ('connector' = 'datagen', " +
                "'rows-per-second' = '10', " +
                "'fields.col1.length' = '5', " +
                "'fields.col2.min' = '1001', " +
                "'fields.col2.max' = '1099'" +
                ")");

        //sink
        tableEnv.executeSql("create table sinkTable(sinkCol1 String, sinkCol2 bigint, sinkCol3 timestamp) with (" +
                "'connector' = 'filesystem', " +
                "'path' = 'E:\\workspaces_sdses\\bigdata-flink\\src\\main\\resources\\sinkData\\file', " +
                "'format' = 'csv'" +
//                "'field-delimiter' = '|'" +
                ")");
        //sink
        tableEnv.executeSql("insert into sinkTable select col1, col2, col3 from sourceTable");


    }
}
